package com.atbeijing.app;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.atbeijing.constants.GmallConstants;
import com.atbeijing.utils.MyKafkaSender;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import java.net.InetSocketAddress;
import java.util.List;

/**
 * 监控mysql 将数据写入kafka
 */
public class CanalClient {
    public static void main(String[] args) throws InvalidProtocolBufferException {
        //1.获取canal连接对象example为canal配置文件所在目录
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop202", 11111), "example", "", "");
        //2.获取连接
        canalConnector.connect();
        //3.指定canal要监控的数据库
        canalConnector.subscribe("gmall20.*");

        while (true) {
            //4.获取Message
            Message message = canalConnector.get(100);

            List<CanalEntry.Entry> entries = message.getEntries();
            if (entries.size() <= 0) {
                System.out.println("没有数据，休息一会");
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                for (CanalEntry.Entry entry : entries) {

                    //TODO 获取表名
                    String tableName = entry.getHeader().getTableName();

                    //Entry类型
                    CanalEntry.EntryType entryType = entry.getEntryType();

                    //判断entryType是否为ROWDATA
                    if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
                        //获取序列化了的数据
                        ByteString storeValue = entry.getStoreValue();

                        //反序列化
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);

                        //TODO 获取事件类型
                        CanalEntry.EventType eventType = rowChange.getEventType();

                        //TODO 获取具体的数据
                        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();

                        //TODO 根据条件获取数据
                        handler(tableName, eventType, rowDataList);
                    }
                }
            }

        }

    }

    private static void handler(String tableName, CanalEntry.EventType eventType, List<CanalEntry.RowData> rowDataList) {
        //获取订单表的新增数据
        if ("order_info".equals(tableName) && CanalEntry.EventType.INSERT.equals(eventType)) {
            sendToKafka(rowDataList,GmallConstants.KAFKA_TOPIC_ORDER);
        }
        //获取订单表的新增数据
        if ("order_detail".equals(tableName) && CanalEntry.EventType.INSERT.equals(eventType)) {
            sendToKafka(rowDataList,GmallConstants.KAFKA_TOPIC_ORDER_DETAIL);
        }
        //获取用户新增及变化数据
        if ("user_info".equals(tableName) && CanalEntry.EventType.INSERT.equals(eventType)|| CanalEntry.EventType.UPDATE.equals(eventType)) {
            sendToKafka(rowDataList,GmallConstants.KAFKA_TOPIC_USER);
        }
    }


    private static void sendToKafka(List<CanalEntry.RowData> rowDataList,String topic) {
        //遍历每行数据
        for (CanalEntry.RowData rowData : rowDataList) {
            //获取一行数据的列名和值的集合
            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
            //将每行数据转成json对象
            JSONObject jsonObject = new JSONObject();
            for (CanalEntry.Column column : afterColumnsList) {
                jsonObject.put(column.getName(), column.getValue());
            }
            System.out.println(jsonObject.toString());
            //将消息发送到kafka
            MyKafkaSender.send(topic, jsonObject.toString());
        }
    }

}
